/*
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package com.analysys.presto.connector.hbase.optimization;

import com.analysys.presto.connector.hbase.meta.HBaseTableHandle;
import com.analysys.presto.connector.hbase.meta.HBaseTableLayoutHandle;
import com.facebook.airlift.log.Logger;
import com.facebook.presto.spi.ColumnHandle;
import com.facebook.presto.spi.TableHandle;
import com.facebook.presto.spi.adaptive.CatalogEngine;
import com.facebook.presto.spi.adaptive.ConnectorAdaptiveOptimizer;
import com.facebook.presto.spi.adaptive.OptimizerType;
import com.facebook.presto.spi.adaptive.metadata.presto.AdaptivePrestoObject;
import com.facebook.presto.spi.plan.FilterNode;
import com.facebook.presto.spi.plan.PlanNode;
import com.facebook.presto.spi.plan.PlanNodeIdAllocator;
import com.facebook.presto.spi.plan.TableScanNode;
import com.facebook.presto.spi.predicate.Domain;
import com.facebook.presto.spi.predicate.Range;
import com.facebook.presto.spi.predicate.TupleDomain;
import com.facebook.presto.spi.relation.VariableReferenceExpression;
import com.facebook.presto.spi.type.BigintType;
import com.facebook.presto.spi.type.IntegerType;
import com.facebook.presto.spi.type.Type;
import com.facebook.presto.spi.type.VarcharType;
import com.google.common.base.Strings;
import io.airlift.slice.Slice;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Optional;

import static com.google.common.base.Preconditions.checkArgument;
import static com.google.common.base.Preconditions.checkState;
import static java.util.Objects.requireNonNull;

/**
 * @author luo
 *
 */
public class HBaseMatchRowkeyPushdown
        implements ConnectorAdaptiveOptimizer
{
    private static final Logger log = Logger.get(HBaseMatchRowkeyPushdown.class);

    @Override
    public CatalogEngine getConnectorCatalog()
    {
        return CatalogEngine.HBASE;
    }

    @Override
    public OptimizerType getOptimizerType()
    {
        return OptimizerType.MATCHROWKEY;
    }

    @Override
    public PlanNode adaptiveOptimize(PlanNode node, TableHandle oldTableHandle, PlanNodeIdAllocator idAllocator, AdaptivePrestoObject adaptiveObject)
    {
        requireNonNull(node, "planNode is null");
        requireNonNull(oldTableHandle, "oldTableHandle is null");
        requireNonNull(idAllocator, "planNodeIdAllocator is null");
        requireNonNull(adaptiveObject, "adaptiveObject is null");

        if (node instanceof FilterNode) {
            TableScanNode oldTableScanNode = (TableScanNode) ((FilterNode) node).getSource();
            TableScanNode newTableScanNode = replaceTableScanNode(oldTableScanNode, oldTableHandle, idAllocator, adaptiveObject);
            if (oldTableScanNode == newTableScanNode) {
                return node;
            }
            // create newFilterNode and return
            return new FilterNode(idAllocator.getNextId(), newTableScanNode, ((FilterNode) node).getPredicate());
        }
        return node;
    }

    private TableScanNode replaceTableScanNode(
            TableScanNode oldTableScanNode,
            TableHandle oldTableHandle,
            PlanNodeIdAllocator idAllocator,
            AdaptivePrestoObject adaptiveObject)
    {
        // get adaptive object.
        TableHandle storedTableHandle = adaptiveObject.getTableHandle();
        Map<String, ColumnHandle> storedColumnHandle = adaptiveObject.getColumnHandles();

        //TODO: tupleDomain is null?
        if (!oldTableHandle.getLayout().isPresent()) {
            return oldTableScanNode;
        }

        // new tupleDomain and replace the old.

        // 1. getOldTupleDomain
        // 2. oldDomains: <oldColumnHandle, domain>
        // ==> newDomains: <HBaseColumnHandle, domain>
        // 3. create newTupleDomain
        TupleDomain<ColumnHandle> oldTupleDomain = oldTableHandle.getLayout().get().getTupleDomain();

        Map<ColumnHandle, Domain> oldDomains = oldTupleDomain.getDomains().get();

        Map<ColumnHandle, Domain> newDomains = new HashMap<>();

        oldDomains.forEach((oldHandle, oldDomain) -> newDomains.put(storedColumnHandle.get(oldHandle.getColumnName()), oldDomain));

        TupleDomain<ColumnHandle> newTupleDomain = TupleDomain.withColumnDomains(newDomains);

        // create startEndKey
        StartEndKey startEndKey = createNewStartEndKey(newDomains, adaptiveObject);

        // create newTableLayoutHandle
        HBaseTableLayoutHandle newTableLayoutHandle =
                new HBaseTableLayoutHandle((HBaseTableHandle) storedTableHandle.getConnectorHandle(), newTupleDomain, startEndKey);

        // create newTableHandle
        TableHandle newTableHandle = new TableHandle(
                storedTableHandle.getConnectorId(),
                storedTableHandle.getConnectorHandle(),
                storedTableHandle.getTransaction(),
                Optional.of(newTableLayoutHandle));

        // create newAssignments
        Map<VariableReferenceExpression, ColumnHandle> newAssignments = new HashMap<>();

        oldTableScanNode.getOutputVariables()
                .forEach(expression -> newAssignments.put(expression, storedColumnHandle.get(expression.getName())));

        // create newTableScanNode
        TableScanNode newTableScanNode = new TableScanNode(
                idAllocator.getNextId(),
                newTableHandle,
                oldTableScanNode.getOutputVariables(),
                newAssignments,
                //TODO: constraint needs to change columnHandle.
                oldTableScanNode.getCurrentConstraint(),
                oldTableScanNode.getEnforcedConstraint()
        );
        return newTableScanNode;
    }

    private StartEndKey createNewStartEndKey(Map<ColumnHandle, Domain> domains, AdaptivePrestoObject adaptiveObject)
    {
        if (domains == null || domains.isEmpty()) {
            log.info("Error: Adaptive Connector Optimizer -- HBaseMatchRowkeyPushDown, domainMap is null.");
            return StartEndKey.ofNull();
        }

        String hbaseRowkey = adaptiveObject.getPrestoTable().getHbaseRowkey();

        if (hbaseRowkey == null) {
            log.info("Error: Adaptive Connector Optimizer -- HBaseMatchRowkeyPushDown, hbaseRowkey is null.");
            return StartEndKey.ofNull();
        }

        // rowkey: salt|indexColumn-length|primaryKey-length
        //      salt|cc-6|pk-5 => 000012312345
        String[] patternParts = hbaseRowkey.split("\\|");

        String[] columnNameAndLength = patternParts[1].split("-");
        String columnName = columnNameAndLength[0];
        int columnLength = Integer.parseInt(columnNameAndLength[1]);

        String[] pkNameAndLength = patternParts[2].split("-");
        int suffixLength = Integer.parseInt(pkNameAndLength[1]);

        for (Map.Entry<ColumnHandle, Domain> columnDomain : domains.entrySet()) {
            ColumnHandle column = columnDomain.getKey();
            Domain domain = columnDomain.getValue();

            if (columnName.equals(column.getColumnName())) {
                StartEndKey noSuffixRowkey = toStartAndEnd(domain, column.getColumnType(), columnLength);
                return appendSuffix(noSuffixRowkey, suffixLength);
            }
        }

        log.info("Error: Adaptive Connector Optimizer -- HBaseMatchRowkeyPushDown, rowkeyPattern can't match domain column.");
        return StartEndKey.ofNull();
    }

    private StartEndKey toStartAndEnd(Domain domain, Type type, int columnLength)
    {
        checkArgument(domain.getType().isOrderable(), "Domain type must be orderable");

        List<Object> singleValues = new ArrayList<>();
        List<Object> greaterValues = new ArrayList<>();
        List<Object> lessValues = new ArrayList<>();

        for (Range range : domain.getValues().getRanges().getOrderedRanges()) {
            checkState(!range.isAll()); // Already checked
            if (range.isSingleValue()) {
                singleValues.add(range.getLow().getValue());
            }
            else {
                if (!range.getLow().isLowerUnbounded()) {
                    switch (range.getLow().getBound()) {
                        case ABOVE:
                            greaterValues.add(range.getLow().getValue());
                            break;
                        case EXACTLY:
                            greaterValues.add(range.getLow().getValue());
                            break;
                        case BELOW:
                            throw new IllegalArgumentException("Low marker should never use BELOW bound");
                        default:
                            throw new AssertionError("Unhandled bound: " + range.getLow().getBound());
                    }
                }
                if (!range.getHigh().isUpperUnbounded()) {
                    switch (range.getHigh().getBound()) {
                        case ABOVE:
                            throw new IllegalArgumentException("High marker should never use ABOVE bound");
                        case EXACTLY:
                            lessValues.add(range.getHigh().getValue());
                            break;
                        case BELOW:
                            lessValues.add(range.getHigh().getValue());
                            break;
                        default:
                            throw new AssertionError("Unhandled bound: " + range.getHigh().getBound());
                    }
                }
            }
        }
        // create startKey and endKey from valueList.
        try {
            if (singleValues.size() >= 1) {
                if (type instanceof BigintType || type instanceof IntegerType) {
                    Long max = findLongMax(singleValues);
                    Long min = findLongMin(singleValues);
                    return new StartEndKey(fillStart(min.toString(), columnLength, '0'),
                            fillStart(max.toString(), columnLength, '0'));
                }
                if (type instanceof VarcharType) {
                    String min = findStringMax(greaterValues);
                    String max = findStringMin(lessValues);
                    return new StartEndKey(fillEnd(min, columnLength, '0'),
                            fillEnd(max, columnLength, '0'));
                }
            }
            else {

                if (type instanceof BigintType || type instanceof IntegerType) {
                    Long min = findLongMax(greaterValues);
                    Long max = findLongMin(lessValues);
                    // min > max => no rowkey
                    if (min > max) {
                        return StartEndKey.ofNull();
                    }

                    // 两侧都无界 (负无穷,正无穷) => no rowkey
                    if (max == Long.MAX_VALUE && min == Long.MIN_VALUE) {
                        return StartEndKey.ofNull();
                    }

                    // 至少一侧有界时
                    // (有界, 无界)
                    if (max == Long.MAX_VALUE) {
                        return new StartEndKey(
                                fillStart(min.toString(), columnLength, '0'),
                                fillStart("", columnLength, '|'));
                    }
                    // (无界, 有界)
                    if (min == Long.MIN_VALUE) {
                        return new StartEndKey(
                                fillStart("", columnLength, '0'),
                                fillStart(max.toString(), columnLength, '0'));
                    }
                    // (有界, 有界)
                    else {
                        return new StartEndKey(
                                fillStart(min.toString(), columnLength, '0'),
                                fillStart(max.toString(), columnLength, '0'));
                    }
                }
                if (type instanceof VarcharType) {
                    String min = findStringMax(greaterValues);
                    String max = findStringMin(lessValues);

                    // 两侧都无界 (负无穷,正无穷) => no rowkey
                    if (max == null && min == null) {
                        return StartEndKey.ofNull();
                    }

                    // 至少一侧有界时
                    // 让无界侧有界：  正无穷 => "|||||||||||||"  负无穷 => "00000000000000"
                    String MAX_VALUE = fillEnd("", columnLength, '|');
                    String MIN_VALUE = fillEnd("", columnLength, '0');
                    if (min == null) {
                        min = MIN_VALUE;
                    }
                    if (max == null) {
                        max = MAX_VALUE;
                    }

                    // min > max => no rowkey
                    if (min.compareTo(max) > 0) {
                        return StartEndKey.ofNull();
                    }

                    // (有界, 无界)
                    if (max.equals(MAX_VALUE)) {
                        return new StartEndKey(
                                fillEnd(min, columnLength, '0'),
                                MAX_VALUE);
                    }
                    // (无界, 有界)
                    if (min.equals(MIN_VALUE)) {
                        return new StartEndKey(
                                MIN_VALUE,
                                fillEnd(max, columnLength, '0'));
                    }
                    // (有界, 有界)
                    else {
                        return new StartEndKey(
                                fillEnd(min, columnLength, '0'),
                                fillEnd(max, columnLength, '0'));
                    }
                }
            }
        }
        catch (Exception ex) {
            log.info("Error: Adaptive Connector Optimizer -- HBaseMatchRowkeyPushDown, " +
                            "range transform rowkey error. Type: %s, StackTrace: %s",
                    type.toString(), ex.getStackTrace());
            return StartEndKey.ofNull();
        }
        return StartEndKey.ofNull();
    }

    /**
     * append suffix to startEndKey.
     */
    private StartEndKey appendSuffix(StartEndKey noSuffixRowkey, int suffixLength)
    {
        if (noSuffixRowkey.isNull()) {
            return StartEndKey.ofNull();
        }
        return new StartEndKey(noSuffixRowkey.getStart() + fillEnd("", suffixLength, '0'),
                noSuffixRowkey.getEnd() + fillEnd("", suffixLength, '|'));
    }

    /**
     * fill the string
     * fillStart("ab", 5, '2')  =>  "222ab"
     */
    private String fillStart(String str, int len, char ch)
    {
        return Strings.padStart(str, len, ch);
    }

    /**
     * fill the string
     * fillEnd("ab", 5, '2')  =>  "ab222"
     */
    private String fillEnd(String str, int len, char ch)
    {
        return Strings.padEnd(str, len, ch);
    }

    private String transToString(Slice slice)
    {
        byte byteArray[] = (byte[]) slice.getBase();
        StringBuilder str = new StringBuilder();
        for (int i = 0; i < slice.length(); i++) {
            str.append((char) byteArray[i]);
        }
        return str.toString();
    }

    /*-------------------------------compare methods--------------------------------*/
    //TODO: these methods are fool! we can use guava Ordering.natural.max(List)!

    private Long findLongMax(List<Object> greaterValues)
    {
        Long x = Long.MIN_VALUE;
        if (greaterValues == null || greaterValues.isEmpty()) {
            return x;
        }
        for (Object obj : greaterValues) {
            Long y = (Long) obj;
            if (y > x) {
                x = y;
            }
        }
        return x;
    }

    private Long findLongMin(List<Object> lessValues)
    {
        Long x = Long.MAX_VALUE;
        if (lessValues == null || lessValues.isEmpty()) {
            return x;
        }
        for (Object obj : lessValues) {
            Long y = (Long) obj;
            if (y < x) {
                x = y;
            }
        }
        return x;
    }

    private String findStringMax(List<Object> greaterValues)
    {
        if (greaterValues == null || greaterValues.isEmpty()) {
            return null;
        }
        String x = transToString((Slice) greaterValues.get(0));
        for (Object obj : greaterValues) {
            String y = transToString((Slice) obj);
            if (y.compareTo(x) > 0) {
                x = y;
            }
        }
        return x;
    }

    private String findStringMin(List<Object> lessValues)
    {
        if (lessValues == null || lessValues.isEmpty()) {
            return null;
        }
        String x = transToString((Slice) lessValues.get(0));
        for (Object obj : lessValues) {
            String y = transToString((Slice) obj);
            if (y.compareTo(x) < 0) {
                x = y;
            }
        }
        return x;
    }
}


